package cn.iocoder.yudao.module.forward.mqtt.dataSource;

import cn.iocoder.yudao.module.forward.dal.dataobject.datasourceinfo.DataSourceInfoDO;
import cn.iocoder.yudao.module.forward.mqtt.*;
import cn.iocoder.yudao.module.forward.mqtt.MessageHandler;
import cn.iocoder.yudao.module.forward.service.datasourceinfo.DataSourceInfoService;
import com.alibaba.druid.util.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

/**
 * 信息源mqttInfo
 * topics 存放 key:topic   value:topicType
 * name为信息源名称
 */
@Slf4j
@Service
public class SourceMqttManager extends MqttManager {
    @Resource
    private DataSourceInfoService dataSourceInfoService;
    @Resource(name = "SourceMessageHandler")
    private MessageHandler sourceMessageHandler;

    public SourceMqttManager() {
        tag = "信息源";
    }

    @Override
    public void doConnect(MqttInfo mqttInfo) {
        if (mqttInfo == null) {
            return;
        }
        if (clientMaps.containsKey(mqttInfo.getName())) {
            log.error("内部错误, {}({})已存在", tag, mqttInfo.getName());
            return;
        }
        log.info("mqttInfo = {}", mqttInfo);
        // TODO: 2025/4/22
        //  回调函数类可单独处理，也可统一处理 单独处理时需要重新实现相关方法
//        MessageHandler handler = null;
//        switch (mqttInfo.getName()) {
//            default:
//                handler = new SourceMessageHandler(brokerMqttManager, scopeService, strategyService, eewService, eqrService);
//                break;
//        }
//        if (handler == null) {
//            log.error("{}({})不存在的消息处理器，取消创建", tag, mqttInfo.getName());
//            return;
//        }
        createClient(mqttInfo, new MqttCallBackImpl(this, sourceMessageHandler, tag) {
            @Override
            public void subscribeTopics(Integer id) {
                DataSourceInfoDO info = dataSourceInfoService.getDataSourceInfoBySourceId(id);
                Map<String, String> topics = new HashMap<>();
                if (!StringUtils.isEmpty(info.getTopicEew())) {
                    topics.put(info.getTopicEew(), MqttConstants.TOPIC_TYPE_EEW);
                }
                if (!StringUtils.isEmpty(info.getTopicEqr())) {
                    topics.put(info.getTopicEqr(), MqttConstants.TOPIC_TYPE_EQR);
                }
                if (!StringUtils.isEmpty(info.getTopicAckFirst())) {
                    topics.put(MqttConstants.TOPIC_TYPE_ACK, info.getTopicAckFirst());
                }
                if (!StringUtils.isEmpty(info.getTopicAckSecond())) {
                    topics.put(MqttConstants.TOPIC_TYPE_ACK2, info.getTopicAckSecond());
                }

                if (topics != null && topics.size() > 0) {
                    MqttConnect mqttConnect = getMqttConnect(mqttInfo.getName());
                    for (String topic : topics.keySet()) {
                        if (topic.equals(MqttConstants.TOPIC_TYPE_ACK) || topic.equals(MqttConstants.TOPIC_TYPE_ACK2)) {
                            continue;
                        }
                        mqttConnect.subscribe(topic);
                    }
                    mqttInfo.getTopics().putAll(topics);
                }
            }
        });
    }

//    @Override
//    public MqttInfo subscribe(String name, String topic, String topicType) {
//        MqttInfo mqttInfo = super.subscribe(name, topic, topicType);
//        if (mqttInfo != null) {
//            mqttInfo.getTopics().put(topic, topicType);
//            return mqttInfo;
//        }
//        return null;
//    }
//
//    @Override
//    public MqttInfo unSubscribe(String name, String topic, String topicType) {
//        MqttInfo mqttInfo = super.unSubscribe(name, topic, topicType);
//        if (mqttInfo != null) {
//            mqttInfo.getTopics().remove(topic);
//            return mqttInfo;
//        }
//        return null;
//    }
}
